package olegroshka.src.com.bofa.olegroshka.datasource;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import olegroshka.src.com.bofa.olegroshka.aggregators.Aggregator;

public class GroupByDataSource implements DataSource<Object[]> {

	private final DataSource<Object[]> dataSource;
	private final int groupByColumn;
	private final Aggregator<Object, Object>[] aggregators; 
	private final int aggregatesCount;
	private Map<Object, Object[]> map = new HashMap<Object, Object[]>(); 
	
	public GroupByDataSource(
			DataSource<Object[]> dataSource, 
			int groupByColumn, 
			Aggregator<Object, Object>[] aggregators) {
		this.dataSource = dataSource;
		this.groupByColumn = groupByColumn;
		this.aggregators = aggregators;
		int aggregatesCountTmp = 0;
		for (int i = 0; i < aggregators.length; i++) {
			if( aggregators[i] != null ) {
				aggregatesCountTmp++;
			}			
		}
		aggregatesCount = aggregatesCountTmp;
	}

	public Iterator<Object[]> iterator() {
		return map.values().iterator();
	}
	
	private void aggregateData() {
		Iterator<Object[]> iterator = dataSource.iterator();
		while (iterator.hasNext()) {
			Object[] row = iterator.next();
			Object key = row[groupByColumn];
			Object[] aggregatedRow = map.get(key);
			if( aggregatedRow == null ) {
				aggregatedRow = new Object[aggregatesCount];
				map.put(key, aggregatedRow);
			}
			
			int aggregatesIndex = 0;
			for (int i = 0; i < aggregators.length; i++) {
				Aggregator<Object, Object> aggregator = aggregators[i];
				if( aggregator != null ) {
					Object aggregate = aggregatedRow[aggregatesIndex];
					aggregatedRow[aggregatesIndex] = aggregator.aggregate(aggregate, row[i]);
					++aggregatesIndex;
				} 				
			}
			
		}
	}

	public void begin() {
		dataSource.begin();
		aggregateData();
	}

	public void end() {
		dataSource.end();
		map.clear();
	}

}
